package com.lengxf.flink.source;

import com.lengxf.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class MapDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("s1", 1, 1L),
                new WaterSensor("s1", 1, 11L),
                new WaterSensor("s2", 2, 2L),
                new WaterSensor("s3", 3, 3L)
        );
        //source.map((MapFunction<WaterSensor, String>) value -> value.id)
        //        .print();
        source
                //.filter(e -> e.id.equals("s1"))
                .keyBy(new MyKeySelector())
                .sum("vc")
                //.map(new MyMapFunction())
                .print();
        env.execute();
    }

    /**
     * 返回一个keyedStream
     * key by 不是转换算子 只是对数据进行了重新分区
     * 不能够设置并行度
     * 是对数据分组， 保证相同的key的数据在同一个分区
     * 分区 => 一个子任务 可以理解为一个分区
     *
     * @author Lengxf
     */
    public static class MyKeySelector implements KeySelector<WaterSensor, String> {
        @Override
        public String getKey(WaterSensor value) throws Exception {
            return value.id;
        }
    }

    /**
     * map一进一出
     * 使用return
     *
     * @author Lengxf
     */
    public static class MyMapFunction implements MapFunction<WaterSensor, String> {
        @Override
        public String map(WaterSensor value) throws Exception {
            return value.id;
        }
    }

    /**
     * flatmap一进多出
     * 通过Collector来输出
     * 调用几次就输出几次
     *
     * @author Lengxf
     */
    public static class MyFlatMapFunction implements FlatMapFunction<WaterSensor, String> {
        @Override
        public void flatMap(WaterSensor value, Collector<String> out) throws Exception {
            out.collect(value.id);
        }
    }

}
